package com.cml.domain.service;

import com.alibaba.fastjson.JSONObject;
import com.cml.common.constant.MessageType;
import com.cml.common.constant.NodeStatus;
import com.cml.common.context.ServerContext;
import com.cml.common.dto.entity.SoltMetadata;
import com.cml.common.util.FilePersistUtils;
import com.cml.common.util.SoltMathUtil;
import com.cml.domain.entity.DefaultMessage;
import com.cml.domain.entity.ServerNode;
import com.cml.domain.entity.Solt;
import com.cml.infrastructure.MessageQueueManager;
import com.cml.infrastructure.NetworkManager;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 姓名：陈茂林
 * 时间：2023/11/21 8:30
 * 描述：
 */
public abstract class AbsractSoltManagerService {
    private ServerNode curServerNode;

    private NetworkManager networkManager;

    //考虑到副本转正情况，会出现当前节点多个元数据
    private List<SoltMetadata> curServerNodeSoltMetadata=new ArrayList<>();

    private Map<Integer, Solt> curServerNodeSoltContainer = new ConcurrentHashMap();

    private List<SoltMetadata> slotsAllocation = new ArrayList();

    private WaitGetCurServerNodeSoltMeatadataThread waitGetCurServerNodeSoltMeatadataThread;
    Logger logger = Logger.getLogger(AbsractSoltManagerService.class);

    /**
     * 槽位分配存储文件的名字
     */
    private static final String SLOTS_ALLOCATION_FILENAME = "slots_allocation";
    public AbsractSoltManagerService(ServerContext serverContext,SoltContextManagerService soltContextManagerService){
        this.curServerNode = serverContext.getServerNode();
        this.networkManager=serverContext.getNetworkManager();
    }

    public ServerNode getCurServerNode() {
        return curServerNode;
    }

    public List<SoltMetadata> getCurServerNodeSoltMetadata() {
        return curServerNodeSoltMetadata;
    }

    public void setCurServerNodeSoltMetadata(SoltMetadata curServerNodeSoltMetadata ) {
        this.curServerNodeSoltMetadata.add(curServerNodeSoltMetadata);
    }

    public NetworkManager getNetworkManager() {
        return networkManager;
    }

    public List<SoltMetadata> getSlotsAllocation() {
        return slotsAllocation;
    }

    public Solt getServiceNameSolt(String serviceName){
        return curServerNodeSoltContainer.get(SoltMathUtil.mathSolt(serviceName));
    }

    protected void initSoltData(){
        curServerNodeSoltMetadata=new ArrayList<>();
        slotsAllocation=new ArrayList<>();
        curServerNodeSoltContainer=new HashMap<>();
    }





    public void startSoltManagerProcess(){
        try{
            if(this.curServerNode.isLeader()){
                //大于0表示已经存在副本
                if(slotsAllocation.size() == 0){
                    //分配所有的solt
                    this.allocateSolt();
                }

                if(this.persistSlotsAllocation()){
                    logger.info("持久化槽位分配数据到本地磁盘......");
                    //同步到所有follower
                    synSoltMetadataToFollower();
                    //发送所有节点负责的槽位信息
                    sendServerNodeSolt();
                }else{
                    NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
                }
            }else{
                this.initSoltData();
                if(this.curServerNode.isFollower()){
                    waitLeaderSynSoltMeta();
                }
                //todo 开启线程接听
                waitGetCurServerNodeSoltMeatadata();
            }
            logger.info("当前分配槽位为："+this.curServerNodeSoltMetadata);
            this.initSoltMsg();
            logger.info("初始化槽信息完成");
            //监听节点断联，副本转正多个副本情况
            if(waitGetCurServerNodeSoltMeatadataThread == null){
                waitGetCurServerNodeSoltMeatadataThread = new WaitGetCurServerNodeSoltMeatadataThread();
                waitGetCurServerNodeSoltMeatadataThread.start();
            }
        }catch (Exception e){
            logger.error(e.getMessage(),e);
            NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
        }
    }

    protected abstract void allocateSolt();

    public   void synSoltMetadataToFollower() throws Exception {
        synSoltMetadataToFollower(getSoltMetadataSynType());
    }

    protected abstract Integer getSoltMetadataSynType();

    private void synSoltMetadataToFollower(Integer soltMetadataType) throws Exception {
        DefaultMessage defaultMessage = new DefaultMessage();
        String messageJson = JSONObject.toJSONString(this.getSlotsAllocation());
        defaultMessage.setDefaultMessage(messageJson.getBytes());
        defaultMessage.setDefaultMessageType(soltMetadataType);
        this.networkManager.sendAllFollowerServerNodeMessage(defaultMessage);
        logger.info("同步follower槽信息为："+messageJson);
    }

    /**
     * 持久化槽位分配数据到本地磁盘
     */
    private Boolean persistSlotsAllocation() {
        String slotsAllocationJSON = JSONObject.toJSONString(slotsAllocation);
        byte[] slotsAllocationByteArray = slotsAllocationJSON.getBytes();
        return FilePersistUtils.persist(slotsAllocationByteArray, SLOTS_ALLOCATION_FILENAME+this.getClass().getName());
    }

    protected abstract Integer  getSendServerNodeSoltMessageType();

    private void sendServerNodeSolt() throws Exception {
        List<SoltMetadata>  waitSendSoltMetadata = this.slotsAllocation;
        for(SoltMetadata soltMetadata : waitSendSoltMetadata){
            if(this.curServerNode.getId().equals(soltMetadata.getServerNodeId())){
                this.curServerNodeSoltMetadata.add(soltMetadata);
            }else{
                DefaultMessage message = new DefaultMessage();
                message.setDefaultMessageType(getSendServerNodeSoltMessageType());
                String messageJson = JSONObject.toJSONString(soltMetadata);
                message.setDefaultMessage(messageJson.getBytes());
                MessageQueueManager.sendMessage(soltMetadata.getServerNodeId(),message);
            }
        }
    }

    private void waitLeaderSynSoltMeta() throws InterruptedException {
        logger.info("开始等待leader同步槽信息");
        DefaultMessage defaultMessage = (DefaultMessage) MessageQueueManager.takeSoltMetadataMessage();
        String messageJson = new String(defaultMessage.getDefaultMessage());
        this.slotsAllocation = JSONObject.parseArray(messageJson,SoltMetadata.class);
        logger.info("获取到Leader同步的所有槽信息为:" +messageJson);
    }

    protected abstract void waitGetCurServerNodeSoltMeatadata() throws InterruptedException ;

    private void initSoltMsg(){
        for (int i = 0; i < curServerNodeSoltMetadata.size(); i++) {
            SoltMetadata soltMetadata=curServerNodeSoltMetadata.get(i);
            for(int soltNo = soltMetadata.getStartSlot();soltNo<=soltMetadata.getEndSlot();soltNo++){
                Solt solt=new Solt(soltNo,doGetServiceInstanceManagerService());
                if(this.curServerNodeSoltContainer.get(soltNo)== null){
                    this.curServerNodeSoltContainer.put(soltNo,solt);
                }

            }

        }
    }

    protected abstract ServiceInstanceManagerService doGetServiceInstanceManagerService();


    class WaitGetCurServerNodeSoltMeatadataThread extends  Thread{

        @Override
        public void run() {
            while (true){
                try {
                    waitGetCurServerNodeSoltMeatadata();
                    initSoltMsg();
                }catch (Exception e){
                    logger.error(e.getMessage(),e);
                    NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
                }
            }
        }
    }
}
